/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.datadispatch.handler.datadispatch;

import static cn.ac.iie.ulss.datadispatch.handler.datadispatch.DataDispatchHandler.logger;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import javax.servlet.ServletException;
import javax.servlet.ServletInputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.handler.AbstractHandler;

/**
 *
 * @author alexmu
 */
public class DataDispatchHandler extends AbstractHandler {

    //docs shcema
    private static String docsSchemaContent = null;
    private static String docSchemaContent = null;
    private static Schema docsSchema = null;
    private static DatumReader<GenericRecord> docsReader = null;
    /*
     */
    private static Schema docSchema = null;
    private static DatumReader<GenericRecord> docReader = null;

    /*
     * 
     */
    private static DataDispatchHandler dataDispatchHandler = null;
    private static List<String> knownSchemaName = new ArrayList<String>();
    private Map<String, DataDispatcher> docSchema2DispatcherSet = new HashMap<String, DataDispatcher>();
    static Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = Logger.getLogger(DataDispatchHandler.class.getName());
    }

    private DataDispatchHandler() {
    }

    public static DataDispatchHandler getDataDispatchHandler() {
        if (dataDispatchHandler != null) {
            return dataDispatchHandler;
        }
        dataDispatchHandler = new DataDispatchHandler();

        try {
            Protocol docsProtocol = Protocol.parse(new File("docs.json"));
            docsSchema = docsProtocol.getType("docs");
            docsReader = new GenericDatumReader<GenericRecord>(docsSchema);
            logger.info("parse schema docs successfully");
            if (docsSchema == null) {
                logger.error("schema docs is not found in metadb,please check metadb to ensure that docs schema exist");
                dataDispatchHandler = null;
            }
            FileInputStream docfis = new FileInputStream("doc.json");
            docSchemaContent = docfis.toString();
            docfis.close();
            logger.info("parsing data schemas finished");
        } catch (Exception ex) {
            logger.error("constructing data dispath handler failed for " + ex.getMessage(), ex);
            dataDispatchHandler = null;
        }

        try {
            Protocol docProtocol = Protocol.parse(new File("doc.json"));
            docSchema = docProtocol.getType("doc");
            docReader = new GenericDatumReader<GenericRecord>(docSchema);
            logger.info("parse schema doc successfully");
        } catch (Exception ex) {
            logger.error("constructing data dispath handler failed for " + ex.getMessage(), ex);
            dataDispatchHandler = null;
        }
        return dataDispatchHandler;
    }

    public static String getDocSchemaContent(String pSchemaName) {
        DataDispatcher dataDispatcher = dataDispatchHandler.docSchema2DispatcherSet.get(pSchemaName);

        if (dataDispatcher == null) {
            return null;
        } else {
            return dataDispatcher.getDocSchemaContent();
        }
    }

    @Override
    public void handle(String string, Request baseRequest, HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse) throws IOException, ServletException {
        long startTime = System.nanoTime();
        String reqID = String.valueOf(startTime);
        try {
            baseRequest.setHandled(true);

            String remoteHost = baseRequest.getRemoteAddr();
            int remotePort = baseRequest.getRemotePort();
            logger.info("receive request from " + remoteHost + ":" + remotePort + " and assigned id " + reqID);

            //retrive data
            ServletInputStream servletInputStream = null;
            byte[] req = null;
            try {
                logger.debug("req " + reqID + ":retriving bussisness data for request  ...");
                servletInputStream = baseRequest.getInputStream();
                ByteArrayOutputStream out = new ByteArrayOutputStream();
                byte[] b = new byte[4096];
                int i = 0;
                while ((i = servletInputStream.read(b, 0, 4096)) > 0) {
                    out.write(b, 0, i);
                }
                req = out.toByteArray();
                logger.debug("req " + reqID + ":length of bussisness data is " + req.length);
            } catch (Exception ex) {
                String errInfo = "req " + reqID + ":retrive bussiness data unsuccessfully for " + ex.getMessage();
                logger.error(errInfo, ex);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + errInfo);
                return;
            } finally {
                try {
                    servletInputStream.close();
                } catch (Exception ex) {
                }
            }

            if (req == null || req.length == 0) {
                String warnInfo = "req " + reqID + ":retrive bussiness data unsuccessfully for content is empty";
                logger.error(warnInfo);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + warnInfo);
                return;
            }

            //****************************
            //decode data
            //****************************
            ByteArrayInputStream docsbis = null;
            GenericRecord docsRecord = null;
            try {
                logger.debug("req " + reqID + ":decoding bussisness data ...");
                docsbis = new ByteArrayInputStream(req);//tuning
                BinaryDecoder docsbd = new DecoderFactory().binaryDecoder(docsbis, null);
                docsRecord = new GenericData.Record(docsSchema);
                docsReader.read(docsRecord, docsbd);
                logger.debug("req " + reqID + ":decode bussisness data sccessfully");
            } catch (Exception ex) {
                String errInfo = "req " + reqID + ":decode bussiness data unsuccessfully for " + ex.getMessage();
                logger.error(errInfo, ex);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + errInfo);
                return;
            } finally {
                try {
                    docsbis = null;
                } catch (Exception ex) {
                }
            }

            //check format of docs         
            String docSchemaName = null;
            try {
                docSchemaName = docsRecord.get("doc_schema_name").toString();
                if (docSchemaName == null || docSchemaName.isEmpty()) {
                    String errInfo = "req " + reqID + ":docSchemaName is empty";
                    logger.error(errInfo);
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                    return;
                }
            } catch (Exception ex) {
                String errInfo = "req " + reqID + ":wrong format of bussiness data,can't get docSchemaName";
                logger.error(errInfo, ex);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + errInfo);
                return;
            }
            logger.debug("req " + reqID + ":doc shcema name is " + docSchemaName);

            String doc_desc = null;
            try {
                doc_desc = docsRecord.get("doc_desc").toString();
                if (docSchemaName == null || docSchemaName.isEmpty()) {
                    String errInfo = "req " + reqID + ":doc_desc is empty";
                    logger.error(errInfo);
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                    return;
                }
            } catch (Exception ex) {
                String errInfo = "req " + reqID + ":wrong format of bussiness data,can't get doc_desc";
                logger.error(errInfo, ex);
                httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                httpServletResponse.getWriter().println("-1\n" + errInfo);
                return;
            }
            logger.debug("req " + reqID + ":doc_desc is " + doc_desc);

            try {
                GenericArray docsSet = (GenericData.Array<GenericRecord>) docsRecord.get("doc_set");
                if (docsSet == null) {
                    String errInfo = "req " + reqID + ":wrong format of bussiness data, doc_set is null";
                    logger.error(errInfo);
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                } else if (docsSet.size() <= 0) {
                    String errInfo = "req " + reqID + ":doc_set is empty";
                    logger.error(errInfo);
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                } else {
                    for (Object o : docsSet) {
                        ByteArrayInputStream bis = new ByteArrayInputStream(((ByteBuffer) o).array());
                        BinaryDecoder bd = new DecoderFactory().binaryDecoder(bis, null);
                        GenericRecord record = (GenericRecord) docReader.read(null, bd);
                        logger.info("the record is " + record);
                    }

                }

                try {
                    httpServletResponse.setStatus(HttpServletResponse.SC_OK);
                    httpServletResponse.getWriter().println("0\n");
                } catch (Exception ex) {
                    String errInfo = "req " + reqID + ":internal error for sending data unsuccessfully for " + ex.getMessage();
                    logger.error(errInfo, ex);
                    httpServletResponse.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
                    httpServletResponse.getWriter().println("-1\n" + errInfo);
                }
            } catch (Exception e) {
                logger.error(e, e);
            }
        } catch (Exception ex) {
            String errInfo = "req " + reqID + ":internal error for sending data unsuccessfully for " + ex.getMessage();
            logger.error(errInfo, ex);
            httpServletResponse.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
            httpServletResponse.getWriter().println("-1\n" + errInfo);
        } finally {
            long endTime = System.nanoTime();
            logger.info("dispose req " + reqID + " consume " + (endTime - startTime) / (1000 * 1000) + " ms");
        }
    }
}

    /*
     * 
     *             //knownSchemaName.addAll(ZkUtil.getChildren(RuntimeEnv.zc, "/ulss/dd/data2mq"));

     if (knownSchemaName.isEmpty()) {
     logger.warn("No business data schema in zookeeper");
     }
     for (String datatype : knownSchemaName) {
     //String mq = ZkUtil.readData(RuntimeEnv.zc, "/ulss/dd/data2mq/" + datatype);
     DataDispatcher dataDispatcher = DataDispatcher.getDataDispatcher(datatype, docSchemaContent, "default_data_mq");
     if (dataDispatcher != null) {
     dataDispatchHandler.docSchema2DispatcherSet.put("doc", dataDispatcher);
     logger.info("construct data dispatcher for schema " + datatype + " successfully");
     dataDispatchHandler.docSchema2DispatcherSet.put(datatype, dataDispatcher);
     logger.info("construct data dispatcher for schema " + datatype + " successfully");
     } else {
     dataDispatchHandler = null;
     logger.info("constructing data dispatcher for schema doc failed");
     break;
     }
     }
     * 
     * 
     */
